package cn.ztuo.bitrade.job;

import cn.ztuo.bitrade.constant.SysConstant;
import cn.ztuo.bitrade.entity.*;
import cn.ztuo.bitrade.handler.NettyHandler;
import cn.ztuo.bitrade.util.ContractUtils;
import cn.ztuo.bitrade.util.RedisUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.simp.SimpMessagingTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.math.BigDecimal;
import java.util.*;

@Component
@Slf4j
public class ContractPushJob {
    @Autowired
    private SimpMessagingTemplate messagingTemplate;
    @Autowired
    private NettyHandler nettyHandler;
    private Map<String, List<ContractTrade>> tradesQueue = new HashMap<>();
    private Map<String, List<TradePlate>> plateContractQueue = new HashMap<>();
    private Map<String, List<CoinThumb>> thumbQueue = new HashMap<>();

    @Autowired
    private RedisUtil redisUtil;

    public void addTrades(String symbol, List<ContractTrade> trades) {
        List<ContractTrade> list = tradesQueue.get(symbol);
        if (list == null) {
            list = new ArrayList<>();
            tradesQueue.put(symbol, list);
        }
        synchronized (list) {
            list.addAll(trades);
        }
    }

    public void addPlates(String symbol, TradePlate plate) {
        List<TradePlate> list = plateContractQueue.get(symbol);
        if (list == null) {
            list = new ArrayList<>();
            plateContractQueue.put(symbol, list);
        }
        synchronized (list) {
            list.add(plate);
        }
    }

    public void addThumb(String symbol, CoinThumb thumb) {
        List<CoinThumb> list = thumbQueue.get(symbol);
        if (list == null) {
            list = new ArrayList<>();
            thumbQueue.put(symbol, list);
        }
        synchronized (list) {
            list.add(thumb);
        }
    }


    @Scheduled(fixedRate = 500)
    public void pushTrade() {

        Iterator<Map.Entry<String, List<ContractTrade>>> entryIterator = tradesQueue.entrySet().iterator();
        while (entryIterator.hasNext()) {
            Map.Entry<String, List<ContractTrade>> entry = entryIterator.next();
            String symbol = entry.getKey();
            String key = SysConstant.CONTRACT_PRICE_CONTROL+symbol+":";
            List<ContractTrade> trades = entry.getValue();
            if (trades.size() > 0) {
                synchronized (trades) {
                    //一次最大推送100条成交
                    int maxLength = 100;
                    List<ContractTrade> pushTrades = trades.size() > maxLength ? trades.subList(0, maxLength) : trades;
                    if(redisUtil.hasKey(key)){
                        if (redisUtil.hasKey(key)) {
                            ContractPriceControl priceControl = (ContractPriceControl) redisUtil.get(key);
                            for(ContractTrade trade : pushTrades){
                                trade.setPrice(priceControl.getPrice());
                            }
                        }
                    }
                    messagingTemplate.convertAndSend("/topic/contract-market/trade/" + symbol, pushTrades);
                    trades.clear();
                }
            }
        }
    }


    @Scheduled(fixedRate = 500)
    public void pushPlate() {
        Iterator<Map.Entry<String, List<TradePlate>>> entryIterator = plateContractQueue.entrySet().iterator();
        while (entryIterator.hasNext()) {
            Map.Entry<String, List<TradePlate>> entry = entryIterator.next();
            String symbol = entry.getKey();
            List<TradePlate> plates = entry.getValue();
            if (plates.size() > 0) {
                boolean hasPushAskPlate = false;
                boolean hasPushBidPlate = false;
                synchronized (plates) {
                    log.info("====处理盘口推送信息===");
                    for (TradePlate plate : plates) {
                        if (plate.getDirection() == ContractOrderDirection.BUY && !hasPushBidPlate) {
                            hasPushBidPlate = true;
                        } else if (plate.getDirection() == ContractOrderDirection.SELL && !hasPushAskPlate) {
                            hasPushAskPlate = true;
                        } else {
                            continue;
                        }
                        //websocket推送盘口信息
                        messagingTemplate.convertAndSend("/topic/contract-market/trade-plate/" + symbol, JSON.toJSONString(plate.toJSON(24)));
                        //websocket推送深度信息
                        messagingTemplate.convertAndSend("/topic/contract-market/trade-plate/" + symbol, JSON.toJSONString(plate.toJSON()));
                        //netty推送
                        nettyHandler.handlePlate(symbol, plate);
                    }
                    plates.clear();
                }
            }
        }
    }


    @Scheduled(fixedRate = 1000)
    public void pushThumb() {
        Iterator<Map.Entry<String, List<CoinThumb>>> entryIterator = thumbQueue.entrySet().iterator();
        while (entryIterator.hasNext()) {
            Map.Entry<String, List<CoinThumb>> entry = entryIterator.next();
            String symbol = entry.getKey();
            List<CoinThumb> thumbs = entry.getValue();
            if (thumbs.size() > 0) {
                synchronized (thumbs) {
                    CoinThumb thumb = thumbs.get(thumbs.size() - 1);
                    JSONObject json = new JSONObject();
                    thumb.setContractCoin(json);
                    messagingTemplate.convertAndSend("/topic/contract-market/thumb", thumb);
                    nettyHandler.pushThumb(symbol, thumb);
                    thumbs.clear();
                }
            }
        }
    }

    @Scheduled(fixedRate = 1000)
    public void price() {
        List<ContractCoin> coins = ContractUtils.findBaseSymbols();
        Map<String, BigDecimal> indexPriceMap = ContractUtils.getIndexprice();
//        Map<String, BigDecimal> marketPriceMap = ContractUtils.getMarketprice();
        coins.stream().forEach(coin->{
            JSONObject json = new JSONObject();
            json.put("indexPrice", indexPriceMap.get(coin.getSymbol()));
            json.put("marketPrice", indexPriceMap.get(coin.getSymbol()));
            messagingTemplate.convertAndSend("/topic/contract-market/indexPrice/"+coin.getSymbol(), json);
        });
    }
}
